/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.wx.queueservice;

import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.RESessionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSession;
import cn.ac.iie.di.wx.common.RuntimeEnv;
import cn.ac.iie.di.wx.vo.PicUrlVo;
import java.util.Date;
import java.util.HashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import org.apache.log4j.Logger;

/**
 *
 * @author Administrator
 */
public class SendToQueueThread implements Runnable {

    public String connectionName;
    public String builderName;
    public RESendSession session;
    public String sessonName;
    public String columnName;
    public LinkedBlockingQueue<PicUrlVo> picUrlBlockingQueue;
    static Logger logger = Logger.getLogger(SendToQueueThread.class.getName());

    public SendToQueueThread(String connectionName, String builderName, RESendSession session, String sessonName, String columnName, LinkedBlockingQueue<PicUrlVo> picUrlBlockingQueue) {
        this.connectionName = connectionName;
        this.builderName = builderName;
        this.session = session;
        this.sessonName = sessonName;
        this.columnName = columnName;
        this.picUrlBlockingQueue = picUrlBlockingQueue;
    }

    public SendToQueueThread(RESendSession session, String columnName, LinkedBlockingQueue<PicUrlVo> picUrlBlockingQueue) {
        this.session = session;
        this.columnName = columnName;
        this.picUrlBlockingQueue = picUrlBlockingQueue;
    }

    @Override
    public void run() {
        try {
            session.start();
        } catch (REConnectionException ex) {
            logger.info(ex.getMessage(), ex);
        }
        while (true) {
            try {

                session.setDocSchemaName((String) RuntimeEnv.getParam(RuntimeEnv.DOCSNAME));
                HashMap<String, String> user_desc = new HashMap<>();
//            for (int r = 0; r < 1500; ++r) {//发的包个数
                session.setUserDesc(user_desc);

                try {
                    Thread.sleep(50);
                } catch (InterruptedException ex) {
                }
                String ifsiexist="";
                for (int i = 0; i < 50; i++) {//每个包的大小
                    PicUrlVo vo = picUrlBlockingQueue.poll(100, TimeUnit.MILLISECONDS);
                    if (vo != null) {
                        ifsiexist+=vo.getIsExist()+"|"+vo.getUrl()+"|"+(String)RuntimeEnv.getParam(RuntimeEnv.LOCALHOST)+";";
//                        session.setString(columnName, vo.getUrl());
                        
                    }
                    vo = null;//暗示jvm进行垃圾回收

                }
                if(!ifsiexist.equals("")){
                    session.setStrings(columnName, ifsiexist.split(";"));
                    session.add();
                }
//                logger.info("session add");
//                logger.info("current data size : " + session.getCurrentDataSize());
//                logger.info("current pac num : " + session.getCurrentPacNum());
//                logger.info("record num int current pac : " + session.getRecordNumInCurrentPac());
//                logger.info(new Date().toLocaleString());
                //添加完所有数据，flush即可发送。
                if (session.getRecordNumInCurrentPac() > 0) {
                    session.flush();
                    logger.info("send to queue success");
                } else {
                    logger.info("there is no data in the queue ");
                }

            } catch (RESessionException ex) {
                logger.info(ex.getMessage(), ex);
            } catch (REConnectionException ex) {
                logger.info(ex.getMessage(), ex);
            } catch (InterruptedException ex) {
                logger.info(ex.getMessage(), ex);
            }
        }
    }
}
